package com.poetic.network.flow;

import com.poetic.network.flow.domain.CarWc;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.concurrent.TimeUnit;

/**
 * <pre>
 *  TODO
 * Created by lianghuikun on 2020-09-14.
 * </pre>
 *
 * @author lianghuikun
 */
public class WatermarkTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.getConfig().setAutoWatermarkInterval(200);
        // 测试的时候并行度要设置为1，否则误以为有问题
        env.setParallelism(1);
//        DataStreamSource<String> stream = env.socketTextStream("localhost", 9999);

        DataStreamSource<String> stream = env.addSource(new SourceFunction<String>() {
            private Boolean isRuning = Boolean.TRUE;
            private Integer cnt = 1000;
            private Integer index = 1000;

            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                while (isRuning) {

                    /**
                     * 测试数据：
                     *          1,1,1000
                     *          1,1,4000
                     *          1,1,5000
                     *          1,1,7000
                     *          1,1,9000
                     *          1,1,11000
                     *          1,1,8000
                     *          1,1,11000
                     */
                    ctx.collect("1,1," + cnt);
                    cnt = cnt + index;

                    TimeUnit.SECONDS.sleep(2);
                }

            }

            @Override
            public void cancel() {
                isRuning = Boolean.FALSE;
            }
        });
        SingleOutputStreamOperator<CarWc> dataStream = stream.map(new MapFunction<String, CarWc>() {
            @Override
            public CarWc map(String value) throws Exception {
                if (StringUtils.isBlank(value)) return null;
                String[] arr = value.split("\\,");
                return new CarWc(arr[0], Integer.valueOf(arr[1].trim()), Long.valueOf(arr[2]));
            }
        });
        dataStream.print("data");
        // 添加水印
        SingleOutputStreamOperator<CarWc> watermarkDataStream = dataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<CarWc>(Time.seconds(2)) {
            @Override
            public long extractTimestamp(CarWc element) {
                return element.getTimestamp();
            }

        });


        // 开窗
        WindowedStream<CarWc, String, TimeWindow> windowStream = watermarkDataStream.keyBy(new KeySelector<CarWc, String>() {
            @Override
            public String getKey(CarWc value) throws Exception {
                return value.getId();
            }
        })
                .window(TumblingEventTimeWindows.of(Time.seconds(5)));
        // 允许延迟+侧道输出
        OutputTag<CarWc> outputTag = new OutputTag<CarWc>("lateCarWc") {};
        windowStream = windowStream.allowedLateness(Time.seconds(3))
                .sideOutputLateData(outputTag);
        // 使用apply计算
        SingleOutputStreamOperator<CarWc> result = windowStream.apply(new WindowFunction<CarWc, CarWc, String, TimeWindow>() {
            @Override
            public void apply(String key, TimeWindow window, Iterable<CarWc> input, Collector<CarWc> out) throws Exception {
                String id = null;
                int num = 0;
                StringBuilder sb = new StringBuilder();
                for (CarWc carWc : input) {
                    id = carWc.getId();
                    num += carWc.getNum();
                    sb.append(carWc.getTimestamp() + ",");
                }
                CarWc r = new CarWc();
                r.setId(id);
                r.setNum(num);
                /**
                 * 也可以 通过process或apply的out写出到侧道输出流，可以实现一个流到多个流
                 *         eg: ctx.output(outputTag, "sideout-" + String.valueOf(value));
                 */
                out.collect(r);

                System.out.println("===>窗口开始时间：" + window.getStart() + "===>窗口结束时间：" + window.getEnd());
                System.out.println("窗口数据：" + sb.toString());
            }
        });
        result.print();
        DataStream<CarWc> lateCarWc = result.getSideOutput(outputTag);
        lateCarWc.print("侧道输出流数据");
        /**
         * 测试数据：
         *          1,1,1000
         *          1,1,4000
         *          1,1,5000
         *          1,1,7000
         *          1,1,9000
         *          1,1,11000
         *          1,1,8000
         *          1,1,11000
         */
        env.execute("watermark test");
    }
}

